package com.lxl.hadoop.compress;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;

/**
 * @Author lixilin
 * @Date 2022/1/8 19:34
 * @Description
 */
public class TestCompress {
    public static void main(String[] args) throws Exception{
        // 压缩
//        compress("D:\\bigdata\\input-tmp-data\\etl\\web.log" , "org.apache.hadoop.io.compress.BZip2Codec");
//        compress("D:\\bigdata\\input-tmp-data\\etl\\web.log" , "org.apache.hadoop.io.compress.GzipCodec");
//        compress("D:\\bigdata\\input-tmp-data\\etl\\web.log" , "org.apache.hadoop.io.compress.DefaultCodec");

        // 解压缩
        decompress("D:\\bigdata\\input-tmp-data\\etl\\web.log.bz2");
    }

    private static void decompress(String fileName) throws Exception{
        // 判断是否支持
        CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
        CompressionCodec codec = factory.getCodec(new Path(fileName));
        if (codec == null) {
            System.err.println("cannot find codec for file ");
            return;
        }

        // 获取输入流
        FileInputStream fis = new FileInputStream(new File(fileName));
        CompressionInputStream cis = codec.createInputStream(fis);
        // 获取输出流
        FileOutputStream fos = new FileOutputStream(fileName + ".decode");
        // 对拷
        IOUtils.copyBytes(cis,fos,1024*1024,false);
        // 关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(cis);
        IOUtils.closeStream(fis);

    }

    private static void compress(String fileName, String method) throws Exception{
        // 获取输入流
        FileInputStream is = new FileInputStream(new File(fileName));

        Class clazz = Class.forName(method);
        CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(clazz, new Configuration());
        // 获取输出流
        FileOutputStream fos = new FileOutputStream(new File(fileName + codec.getDefaultExtension()));
        CompressionOutputStream cos = codec.createOutputStream(fos);
        // 流的对拷
        IOUtils.copyBytes(is,cos,1024*1024*5,false);
        // 关闭流
        IOUtils.closeStream(cos);
        IOUtils.closeStream(fos);
        IOUtils.closeStream(is);

    }
}
